package com.lagou.rpc.core;

import com.alibaba.fastjson.JSON;
import com.lagou.rpc.core.config.RpcConfig;
import com.lagou.rpc.core.config.Subscribe;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Component
public class ProviderWatcher {

    private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();


    @Autowired
    private RpcZookeeperClient client;

    /**
     * 保存数据
     */
    private Map<String,PriorityQueue<ProviderNode>> providerMap = new ConcurrentHashMap<String, PriorityQueue<ProviderNode>>();

    /**
     * 缓存数据
     */
    private Map<String,List<ProviderNode>> providerCache = new HashMap<String,List<ProviderNode>>();


    public Map<String,PriorityQueue<ProviderNode>> getProvider() {
        return providerMap;
    }

    @Autowired
    private RpcConfig rpcConfig;

    @PostConstruct
    private void watch() throws Exception {
        if (rpcConfig.getSubscribes() == null){
            return;
        }
        for (Subscribe subscribe : rpcConfig.getSubscribes()) {
            client.getClient().getChildren().usingWatcher(new Watch()).forPath("/rpc/provider/"+subscribe.getName());
            loadNode(subscribe.getName());
        }

        executorService.scheduleAtFixedRate(()->{
            try {
                report();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },5,5, TimeUnit.SECONDS);
    }


    private boolean isWatcher(String serviceName){
       for (Subscribe subscribe : rpcConfig.getSubscribes()){
           if (subscribe.getName().equals(serviceName)){
               return true;
           }
       }
       return false;
    }


    private static String rootPath = "/rpc/provider";

    class Watch implements CuratorWatcher {
        @Override
        public void process(WatchedEvent watchedEvent) throws Exception {
            //是否是子节点变更
            if (watchedEvent.getType().equals(Watcher.Event.EventType.NodeChildrenChanged)){
                //是否是rpc变更了
                if(watchedEvent.getPath().startsWith("/rpc/provider")){
                    String serviceName = watchedEvent.getPath().split("/")[3];
                    loadNode(serviceName);
                }
            }
        }
    }


    private synchronized void loadNode(String serviceName) throws Exception {
        if (isWatcher(serviceName)){
            //构建堆对象
            PriorityQueue<ProviderNode> nodeQueue = new PriorityQueue<ProviderNode>(10, new Comparator<ProviderNode>() {
                @Override
                public int compare(ProviderNode o1, ProviderNode o2) {
                    return o1.getTime().compareTo( o2.getTime()) ;
                }
            });
            List<ProviderNode> providerNodes = new ArrayList<ProviderNode>();
            // 加载节点，并重新构建对象
            String servicePath = rootPath+"/"+serviceName;
            List<String> serviceNodeList = client.getClient().getChildren().forPath(servicePath);
            for (String childrenNode:serviceNodeList) {
                String serviceNodePath = servicePath + "/" + childrenNode;
                byte[] bytes =  client.getClient().getData().forPath(serviceNodePath);
                ProviderNode providerNode = JSON.parseObject(new String(bytes),ProviderNode.class);
                nodeQueue.offer(providerNode);
                providerNodes.add(providerNode);
            }
            providerMap.put(serviceName, nodeQueue);
            providerCache.put(serviceName, providerNodes);
            System.out.println("订阅内容成功");
        }
    }


    /**
     * 更新请求时间和时长
     * @param providerNode
     * @param startTime
     * @param endTime
     */
    public void uploadProvider(ProviderNode providerNode, long startTime, Long endTime){
        providerNode.setRequestTime(new Date());
        providerNode.setTime(endTime-startTime);
    }

    /**
     * 上报请求
     */
    public void report() throws Exception {

        for (List<ProviderNode> providerNodeList: providerCache.values()  ) {
            for (ProviderNode providerNode :providerNodeList) {
                if (System.currentTimeMillis() -  providerNode.getRequestTime().getTime() > 5000){
                    client.getClient().delete().forPath(providerNode.getNodeUrl());
                }else {
                    client.getClient().setData().forPath(providerNode.getNodeUrl(),providerNode.toJson().getBytes());
                }
            }
        }


        for (Subscribe subscribe : rpcConfig.getSubscribes()) {
            client.getClient().getChildren().usingWatcher(new Watch()).forPath("/rpc/provider/"+subscribe.getName());
            loadNode(subscribe.getName());
        }
    }


}
